use std::collections::HashSet;

use clickhouse_rs::ClientHandle;

use crate::base::{get_ch_pool, DiviFactConv, DiviType, Table};

pub struct TrlsBack<'a> {
    divi_type: DiviType,
    tconf: &'a Table,
    tmp_table: String,
}
impl<'a> TrlsBack<'a> {
    pub(super) fn new(divi_type: DiviType, tconf: &'a Table) -> Self {
        let tmp_table = format!("TMP_{}", tconf.table_name.as_str());
        Self {
            divi_type,
            tconf,
            tmp_table,
        }
    }
    async fn collapsing(
        &self,
        client: &mut ClientHandle,
        dfc: &DiviFactConv,
        ins_col_names_str: &str,
        sel_col_names_str: &str,
    ) -> anyhow::Result<()> {
        let bname = self.divi_type.back_table_name();
        let kcn = dfc.key_col_name.as_str();
        let and_where = if let Some(wcn) = dfc.where_col_name.as_deref() {
            let wcv = dfc.where_col_val.as_deref().unwrap();
            format!("and {wcn}='{wcv}'")
        } else {
            String::new()
        };
        let tname = self.tconf.table_name.as_str();
        let sql_str = format!(
            r#"insert into {}({ins_col_names_str},DIVI.DPID,DIVI.EMID,DIVI.DSVL,SIGN)
            select {sel_col_names_str},anyLast(DIVI.DPID),anyLast(DIVI.EMID),anyLast(DIVI.DSVL),-1 from {tname}
            where {kcn} in (select {kcn} from {bname}) {and_where}
            group by DATA_DT,JLID,{kcn}
            HAVING sum(SIGN)>0"#,
            self.tmp_table
        );
        client.execute(&sql_str).await?;
        Ok(())
    }
    async fn insert(
        &self,
        client: &mut ClientHandle,
        dfc: &DiviFactConv,
        ins_col_names_str: &str,
        sel_col_names_str: &str,
    ) -> anyhow::Result<()> {
        let kcn = dfc.key_col_name.as_str();
        let back_dict_name = self.divi_type.back_dict_name();
        let dpids = format!("dictGet('{back_dict_name}','DIVI.DPID',{kcn},DATA_DT)");
        let emids = format!("dictGet('{back_dict_name}','DIVI.EMID',{kcn},DATA_DT)");
        let dsvls = format!("dictGet('{back_dict_name}','DIVI.DSVL',{kcn},DATA_DT)");
        let bname = self.divi_type.join_table_name();
        let and_where = if let Some(wcn) = dfc.where_col_name.as_deref() {
            let wcv = dfc.where_col_val.as_deref().unwrap();
            format!("and {wcn}='{wcv}'")
        } else {
            String::new()
        };
        let tname = self.tconf.table_name.as_str();
        let sql_str = format!(
            r#"insert into {}({ins_col_names_str},DIVI.DPID,DIVI.EMID,DIVI.DSVL,SIGN)
            select {sel_col_names_str},{dpids},{emids},{dsvls},1 from {tname}
            where {kcn} in (select {kcn} from {bname}) {and_where}
            group by DATA_DT,JLID,{kcn}
            HAVING sum(SIGN)>0"#,
            self.tmp_table
        );
        client.execute(&sql_str).await?;
        Ok(())
    }
    pub(super) async fn run(&self) -> anyhow::Result<()> {
        let mut dfc: Option<&DiviFactConv> = None;
        for d in self.tconf.divi_fact_convs.iter() {
            if d.divi_type == self.divi_type {
                dfc = Some(d);
                break;
            }
        }
        if dfc.is_none() {
            return Ok(());
        }
        let dfc = dfc.unwrap();
        let mut col_names: HashSet<&str> = HashSet::new();
        for (k, _v) in self.tconf.columns.iter() {
            col_names.insert(k);
        }
        for (k, _v) in self.tconf.rs_calc_cols.iter() {
            col_names.insert(k);
        }
        let mut key_col_name: Option<&str> = None;
        for (k, v) in self.tconf.db_calc_cols.iter() {
            col_names.insert(k);
            if v == "#INCR" {
                key_col_name = Some(k);
            }
        }
        if key_col_name.is_none() {
            key_col_name=Some("JLID");
            // Err(anyhow::anyhow!(
            //     "导入文件{}定义了divi_fact_convs，但在db_calc_cols没有找到#INCR列",
            //     self.tconf.file_name
            // ))?
        }
        let key_col_name = key_col_name.unwrap();
        let mut ins_col_names_str = String::new();
        let mut sel_col_names_str = String::new();
        for c in col_names.iter() {
            if *c == "SIGN" {
                continue;
            }
            ins_col_names_str.push_str(c);
            ins_col_names_str.push(',');
            if *c == key_col_name || *c == dfc.key_col_name {
                sel_col_names_str.push_str(c);
            } else {
                sel_col_names_str.push_str("anyLast(");
                sel_col_names_str.push_str(c);
                sel_col_names_str.push(')');
            }
            sel_col_names_str.push(',');
        }
        ins_col_names_str.pop();
        sel_col_names_str.pop();
        let mut client = get_ch_pool().get_handle().await?;
        let sql = format!(
            "CREATE TABLE IF NOT EXISTS {} AS {}",
            self.tmp_table, self.tconf.table_name
        );
        client.execute(&sql).await?;
        let sql = format!("TRUNCATE TABLE {}", self.tmp_table);
        client.execute(sql).await?;
        self.collapsing(&mut client, dfc, &ins_col_names_str, &sel_col_names_str)
            .await?;
        self.insert(&mut client, dfc, &ins_col_names_str, &sel_col_names_str)
            .await?;
        let sql = format!(
            "insert into {} select * from {} order by SIGN",
            self.tconf.table_name, self.tmp_table
        );
        client.execute(&sql).await?;
        let sql = format!("DROP TABLE {}", self.tmp_table);
        client.execute(sql).await?;
        Ok(())
    }
}
